package cn.itcast.n8;

import cn.itcast.n2.util.Sleeper;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


/**
 * @author: AirMan
 * @date: 2025/4/21 14:30
 * @description:
 */
@Slf4j(topic = "c.TestPoll_my")
public class TestPoll_my {

    // programming ==> main
    public static void main(String[] args) {
        ThreadPoll threadPoll = new ThreadPoll(2, 2, 2, TimeUnit.SECONDS,((queue, task) -> {
            // 1.死等
//            queue.put(task);
            // 2.超时等待
//            log.debug("超时等待策略...");
//            queue.offer(task, 1, TimeUnit.SECONDS);
            // 3.直接放弃处理
//            log.debug("放弃处理策略...");
            // 4.报异常
//            throw new RuntimeException();
            // 5.自己处理
            task.run();
        }));
        for (int i = 0; i < 10; i++) {
            int j = i;
            threadPoll.execute(() -> {
                log.debug("任务" + j + "开始...");
                Sleeper.sleep(4);
            });
        }
    }


    interface RejectPolicy<T> {
        void reject(BlockQueue<T> queue, T task);
    }

    // 线程池
    static class ThreadPoll {
        // 任务队列（阻塞）
        private BlockQueue<Runnable> taskQueue;
        // cpu 核心数
        private int coreSize;
        // 获取任务时的超时时间
        private long timeout;
        private TimeUnit timeUnit;
        // 线程集合（工人数量）
        HashSet<Worker> workers = new HashSet<>();
        // 拒绝策略
        RejectPolicy rejectPolicy;

        public ThreadPoll(int coreSize, int capcity, int timeout, TimeUnit timeUnit, RejectPolicy<Runnable> rejectPolicy) {
            this.coreSize = coreSize;
            this.timeout = timeout;
            this.timeUnit = timeUnit;
            taskQueue = new BlockQueue<>(capcity);
            this.rejectPolicy = rejectPolicy;
        }

        // 执行任务队列
        private void execute(Runnable task) {
            synchronized (workers) {
                if (workers.size() < coreSize) {
                    // 当任务数没有超过 coreSize 时，直接交给 worker 对象执行
                    Worker worker = new Worker(task);
                    log.debug("新增 worker{}, {}", worker, task);
                    workers.add(worker);
                    worker.start();
                } else {
                    // 如果任务数超过 coreSize 时，加入任务队列暂存
//                    if (!taskQueue.offer(task, 3, TimeUnit.SECONDS)) {
//                        log.debug("添加超时, 放弃添加！");
//                    }
                    taskQueue.tryPut(rejectPolicy, task);
                }
            }
        }


        class Worker extends Thread {
            private Runnable task;

            public Worker(Runnable task) {
                this.task = task;
            }

            @Override
            public void run() {
                while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                    try {
                        log.debug("worker开始执行task...{}", task);
                        task.run();
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    } finally {
                        task = null;
                    }
                }
                synchronized (workers) {
                    log.debug("移除worker...{}", this);
                    workers.remove(this);
                }
            }
        }
    }

    // 阻塞队列
    static class BlockQueue<T> {
        // 阻塞队列的实现方法
        private Deque<T> queue = new ArrayDeque<>();
        // 阻塞队列的容量
        private int captity;
        // 锁
        private ReentrantLock lock = new ReentrantLock();
        // 条件变量
        private Condition fullWaitSet = lock.newCondition();
        private Condition emptyWaitSet = lock.newCondition();

        public BlockQueue(int captity) {
            this.captity = captity;
        }

        // 获取任务队列中的任务数量
        private int size() {
            lock.lock();
            try {
                return queue.size();
            } finally {
                lock.unlock();
            }
        }

        // 阻塞获取
        private T take() {
            lock.lock();
            try {
                while (queue.isEmpty()) {
                    try {
                        log.debug("当前任务队列为空...");
                        emptyWaitSet.await();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                log.debug("获取到task, 执行任务并唤醒fullWaitSet...");
                T task = queue.removeFirst();
                fullWaitSet.signal();
                return task;
            } finally {
                lock.unlock();
            }
        }

        // 阻塞添加
        private void put(T task) {
            lock.lock();
            try {
                while (queue.size() == captity) {
                    try {
                        log.debug("当前任务队列满...{}", queue.size());
                        fullWaitSet.await();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                log.debug("交给任务队列 ：{}", task);
                queue.addLast(task);
                emptyWaitSet.signal();
            } finally {
                lock.unlock();
            }
        }


        // 阻塞获取(带超时时间)
        private T poll(long time, TimeUnit unit) {
            lock.lock();
            try {
                long nanos = unit.toNanos(time);
                while (queue.isEmpty()) {
                    try {
                        if (nanos <= 0) {
                            log.debug("等待任务，等待超时！");
                            return null;
                        }
                        log.debug("当前任务队列为空...");
                        nanos = emptyWaitSet.awaitNanos(nanos);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                log.debug("获取到task, 执行任务并唤醒fullWaitSet...");
                T task = queue.removeFirst();
                fullWaitSet.signal();
                return task;
            } finally {
                lock.unlock();
            }
        }

        // 阻塞添加(带超时时间)
        private boolean offer(T task, long time, TimeUnit unit) {
            lock.lock();
            try {
                long nanos = unit.toNanos(time);
                while (queue.size() == captity) {
                    try {
                        if (nanos <= 0) {
                            return false;
                        }
                        log.debug("当前任务队列满...{}", queue.size());
                        nanos = fullWaitSet.awaitNanos(nanos);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                log.debug("交给任务队列 ：{}", task);
                queue.addLast(task);
                emptyWaitSet.signal();
                return true;
            } finally {
                lock.unlock();
            }
        }

        private void tryPut(RejectPolicy<T> rejectPolicy, T task) {
            lock.lock();
            try {
                // 判断队列是否已满
                if (queue.size() == captity) {
                    log.debug("队列已满, 执行策略...");
                    rejectPolicy.reject(this, task);
                } else {
                    log.debug("交给任务队列 ：{}", task);
                    queue.addLast(task);
                    emptyWaitSet.signal();
                }
            } finally {
                lock.unlock();
            }
        }

    }
}
